package com.atguigu.flink07;

import com.atguigu.beans.Emp;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author Felix
 * @date 2024/2/26
 * 该案例演示了处理函数
 */
public class Flink03_process_function {
    public static void main(String[] args) {
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(2);
        //TODO 2.从指定的网络端口中读取员工数据
        SingleOutputStreamOperator<Emp> empDS = env
                .socketTextStream("hadoop102", 8888)
                .map(
                        new MapFunction<String, Emp>() {
                            @Override
                            public Emp map(String lineStr) throws Exception {
                                String[] fieldArr = lineStr.split(",");
                                return new Emp(Integer.valueOf(fieldArr[0]), fieldArr[1], Integer.valueOf(fieldArr[2]), Long.valueOf(fieldArr[3]));
                            }
                        })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<Emp>forMonotonousTimestamps()
                                .withTimestampAssigner(
                                        new SerializableTimestampAssigner<Emp>() {
                                            @Override
                                            public long extractTimestamp(Emp emp, long recordTimestamp) {
                                                return emp.ts;
                                            }
                                        }
                                )
                );

        /*empDS.process(
                new ProcessFunction<Emp, String>() {
                    @Override
                    public void processElement(Emp value, ProcessFunction<Emp, String>.Context ctx, Collector<String> out) throws Exception {

                    }
                }
        )
        empDS
                .keyBy()
                .process(
                        new KeyedProcessFunction<Tuple, Emp, Object>() {
                        }
                )

        empDS.keyBy().window().process(
                new ProcessWindowFunction<Emp, Object, Tuple, Window>() {
                }
        )

        empDS.windowAll().process(
                new ProcessAllWindowFunction<Emp, Object, Window>() {
                }
        )


        empDS.connect(empDS).process(
                new CoProcessFunction<Emp, Emp, Object>() {
                }
        )

        empDS.keyBy().intervalJoin().between().process(
                new ProcessJoinFuntion<Emp, Object, Object>() {
                }
        )
        */
    }
}
